package com.cryann.mqttserver.config;

import com.cryann.mqttserver.modle.Accessor;
import com.cryann.mqttserver.modle.AccessorStore;
import com.cryann.mqttserver.repository.AccessorRepository;
import com.cryann.mqttserver.service.MqttAccessor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ScheduledFuture;

/**
 * @author: cRyann
 * @date: 2019/11/14 17:44
 * @description: 动态定时器：用于定时检测连接是否超时
 **/
@Configuration
@Slf4j
@Order(10)
public class AsyScheduler implements CommandLineRunner {
    private static ScheduledFuture<?> future;
    private static double X = 1.5;

    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;
    @Autowired
    private MqttAccessor mqttAccessor;
    @Autowired
    AccessorRepository accessorRepository;

    @Override
    public void run(String... args) throws Exception {
        // 启动之后 开启默认轮询
        future = threadPoolTaskScheduler.scheduleWithFixedDelay(new AliveCheckTask(), new Date(), AccessorStore.CHECK_LIVE_INTERVAL_TIME * 1000);
    }

    public void UpdateAliveCheckTask(int ckeckTime) {
        Optional.ofNullable(future).ifPresent(f -> {
            future.cancel(true);
            future = threadPoolTaskScheduler.scheduleWithFixedDelay(new AliveCheckTask(), new Date(), ckeckTime * 1000);
        });
    }


    /**
     * 检查连接是否存活的进程
     **/
    class AliveCheckTask implements Runnable {
        @Override
        public void run() {
            // TODO 获取Accessor
            log.info("{} action", this.getClass().getSimpleName());
            for (Iterator<Map.Entry<String, Accessor>> it = AccessorStore.instance().getMap().entrySet().iterator(); it.hasNext(); ) {
                Map.Entry<String, Accessor> item = it.next();
                Accessor v = item.getValue();
                long intervalSec = (System.currentTimeMillis() - item.getValue().getActiveTime()) / 1000;
                if ( intervalSec > v.getKeepLive() * X ) {
//                    MARK WillFlag处理
                    mqttAccessor.willMessageSend(item.getKey());
//                    MARK CleanSession 处理
//                    如果保持连接的值非零，并且服务端在一点五倍的保持连接时间内没有收到客户端的控制报文，它必须断开客户端的网络连接，认为网络连接已断开 [MQTT-3.1.2-24]。
//                    客户端发送了PINGREQ报文之后，如果在合理的时间内仍没有收到PINGRESP报文，它应该关闭到服务端的网络连接。
                    v.setOnLine(false);
                    if ( v.getCleanSession() ) {
                        mqttAccessor.delAccessor(v.getChannel());
                    }

                }
            }
        }
    }

}
